package com.shujia.flink.tf;

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Demo8SideOut {
    public static void main(String[] args) throws Exception {
        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        DataStream<String> studentDS = env.socketTextStream("master", 8888);

        // 这需要是一个匿名的内部类，以便我们分析类型
        OutputTag<String> man = new OutputTag<String>("男"){};
        OutputTag<String> female = new OutputTag<String>("女"){};

        /*
         * process算子可以代替map  flatMap filter的功能
         */

        SingleOutputStreamOperator<Object> processDS = studentDS.process(new ProcessFunction<String, Object>() {
            /**
             * processElement:和map算子的map方法类似，一条数据执行一次
             * @param student：一行数据
             * @param context： 上下文对象，用于获取flink的执行环境，以及旁路输出
             * @param collector：用于将数据发送到下游
             */
            @Override
            public void processElement(String student,
                                       ProcessFunction<String, Object>.Context context,
                                       Collector<Object> collector) throws Exception {

                //取出姓名
                String sex = student.split("\\|")[3];

                if ("男".equals(sex)) {
                    //将数据发送到下游，打上标记
                    context.output(man, "man:" + student);
                } else {
                    context.output(female, "female:" + student);
                }

            }
        });

        //通过标记获取连个流
        DataStream<String> manDS = processDS.getSideOutput(man);
        manDS.print();

        DataStream<String> femaleDS = processDS.getSideOutput(female);
        femaleDS.print();

        env.execute();

    }
}
